﻿<#
.DESCRIPTION
    Creates the required tables in the target SQL Server database for the DataCollector operations. 
    Also creates the required settings in the configuration database (connection string and collection interval).
.PARAMETER environmentShortName
    The short name of the environment that contains the configuration.
.PARAMETER serverName
    The name of the server that contains the SQL Server instance. (Should also work with named instance but untested.
.PARAMETER databaseName
    The name of the database that will contain the DataCollector objects.
.PARAMETER schemaName
    The name of the schema within the database that will contain the DataCollector objects.
.PARAMETER collectionIntervalInSeconds
    The number of seconds between each CDC data collections.
.NOTES
    Author: Mathieu Isabel   
#>
Function Set-DataCollectorCDCConfiguration($environmentShortName="",$serverName="",$databaseName="",$schemaName="",$collectionIntervalInSeconds=15)
{
    Import-Module -DisableNameChecking -Force -Name ..\ConfigurationDatabase\libConfigurationDatabase.psm1
    Import-Module -DisableNameChecking -Force -Name ..\ServiceBus\libServiceBus.psm1

    New-ConfigurationSetting -environmentShortName $environmentShortName -scope Global -name "DataCollectorDatabaseConnectionString" -value "Data Source=$serverName;Initial Catalog=$databaseName;Integrated Security=True;MultipleActiveResultSets=True"
    New-ConfigurationSetting -environmentShortName $environmentShortName -scope DataCollector -name "DataCollectorCollectionIntervalInSeconds" -value $collectionIntervalInSeconds

    $sqlConn = new-object -TypeName ("Data.SqlClient.SqlConnection") -ArgumentList (Get-ConfigurationSetting -environmentShortName $environmentShortName -scope "Global"  -name "DataCollectorDatabaseConnectionString").Value

    if($sqlConn.State -ne "Open")
    {
        $sqlConn.Open()
    } 

    $varcharDataType=[System.Data.SqlDbType]::VarChar
    $dateTimeDataType=[System.Data.SqlDbType]::DateTime
    $smallIntDataType=[System.Data.SqlDbType]::SmallInt
    $intDataType=[System.Data.SqlDbType]::Int
    $bigIntDataType=[System.Data.SqlDbType]::BigInt

    $createTrackerTableQuery="CREATE TABLE [$schemaName].[CDCTrackedTables](
	                            [ServerName] [varchar](128) NOT NULL,
	                            [DatabaseName] [varchar](128) NOT NULL,
	                            [SchemaName] [varchar](128) NOT NULL,
	                            [TableName] [varchar](128) NOT NULL,
	                            [LastLSN] [binary](10) NULL,
	                            [CollectionEnabled] [bit] NULL,
                             CONSTRAINT [PK_CDCTrackedTables] PRIMARY KEY CLUSTERED 
                            (
	                            [ServerName] ASC,
	                            [DatabaseName] ASC,
	                            [SchemaName] ASC,
	                            [TableName] ASC
                            )WITH (PAD_INDEX = OFF, STATISTICS_NORECOMPUTE = OFF, IGNORE_DUP_KEY = OFF, ALLOW_ROW_LOCKS = ON, ALLOW_PAGE_LOCKS = ON) ON [PRIMARY]
                            ) ON [PRIMARY]"


    $createTrackerTableCommand = New-Object -TypeName Data.SqlClient.SqlCommand -ArgumentList $createTrackerTableQuery,$sqlConn

    $createTrackerTableCommand.ExecuteNonQuery()


    $createTrackerTableQuery="CREATE TABLE [$schemaName].[CDCTrackedTableColumns](
	                            [ServerName] [varchar](128) NOT NULL,
	                            [DatabaseName] [varchar](128) NOT NULL,
	                            [SchemaName] [varchar](128) NOT NULL,
	                            [TableName] [varchar](128) NOT NULL,
	                            [ColumnName] [varchar](128) NOT NULL,
	                            [TargetReferenceColumnName] [varchar](128) NOT NULL,
	                            [TargetTableName] [varchar](128) NULL,
	                            [TargetKeyColumnName] [varchar](128) NULL,
                             CONSTRAINT [PK_CDCTrackedTableColumns] PRIMARY KEY CLUSTERED 
                            (
	                            [ServerName] ASC,
	                            [DatabaseName] ASC,
	                            [SchemaName] ASC,
	                            [TableName] ASC,
	                            [ColumnName] ASC,
	                            [TargetReferenceColumnName] ASC
                            )WITH (PAD_INDEX = OFF, STATISTICS_NORECOMPUTE = OFF, IGNORE_DUP_KEY = OFF, ALLOW_ROW_LOCKS = ON, ALLOW_PAGE_LOCKS = ON) ON [PRIMARY]
                            ) ON [PRIMARY]"

    $createTrackerTableCommand.ExecuteNonQuery()
    
    $sqlConn.Close()
}

<#
.DESCRIPTION
    Collects the CDC change records on the configured servers, databases, tables and columns.
.PARAMETER environmentShortName
    The short name of the environment that contains the configuration.
.PARAMETER serverName
    The name of the server that contains the SQL Server instance for which the changes will be collected.
.PARAMETER databaseName
    The name of the database for which the changes will be collected.
.PARAMETER schemaName
    The name of the schema that contains the tables for which the changes will be collected.
.PARAMETER cdcTrackedTables
    The list of tables for which the changes will be collected.
.NOTES
    Author: Mathieu Isabel   
#>
Function Get-DataCollectorCDCDatabaseChange($environmentShortName="",$serverName="",$databaseName="",$cdcTrackedTables=$null)
{
    
    Import-Module -DisableNameChecking -Force -Name ..\ConfigurationDatabase\libConfigurationDatabase.psm1
    Import-Module -DisableNameChecking -Force -Name ..\ServiceBus\libServiceBus.psm1

    $dataCollectorConnection = new-object -TypeName ("Data.SqlClient.SqlConnection") -ArgumentList (Get-ConfigurationSetting -environmentShortName $environmentShortName -scope "Global"  -name "DataCollectorDatabaseConnectionString").Value
    $cdcConnection = new-object -TypeName ("Data.SqlClient.SqlConnection") -ArgumentList "Data Source=$serverName;Initial Catalog=$databaseName;Integrated Security=True;MultipleActiveResultSets=True"

    if($dataCollectorConnection.State -ne "Open")
    {
        $dataCollectorConnection.Open()
    }

    if($cdcConnection.State -ne "Open")
    {
        $cdcConnection.Open()
    }

    $varcharDataType=[System.Data.SqlDbType]::VarChar
    $binaryDataType=[System.Data.SqlDbType]::Binary
        
    foreach($cdcTrackedTable in $cdcTrackedTables)
    {
        $tableTrackedColumns=Get-DataCollectorCDCTrackedTableColumn -serverName $cdcTrackedTable.ServerName -databaseName $cdcTrackedTable.DatabaseName -schemaName $cdcTrackedTable.SchemaName -tableName $cdcTrackedTable.TableName
        $cdcMaxLSNQuery="SELECT sys.fn_cdc_get_max_lsn()"

        $cdcMaxLSNCommand = New-Object -TypeName Data.SqlClient.SqlCommand -ArgumentList $cdcMaxLSNQuery,$cdcConnection

        $cdcChangesQuery=Get-DataCollectorCDCQuery -dataCollectorTrackedTable $cdcTrackedTable -dataCollectorTrackedColumns $tableTrackedColumns

        <#$cdcChangesQuery="SELECT *,sys.fn_cdc_map_lsn_to_time(__`$start_lsn) AS _CDC_ChangeTime
                        FROM  [cdc].[fn_cdc_get_all_changes_" + $cdcTrackedTable.SchemaName + "_" + $cdcTrackedTable.TableName + "](@LastLSN, @MaxLSN,N'all')
                        WHERE [__`$start_lsn] !=@LastLSN"
                        #>


        $cdcChangesCommand = New-Object -TypeName Data.SqlClient.SqlCommand -ArgumentList $cdcChangesQuery,$cdcConnection

        $lastLSN = New-Object -TypeName System.Data.SqlClient.SqlParameter -ArgumentList "@LastLSN", $binaryDataType
        $maxLSN = New-Object -TypeName System.Data.SqlClient.SqlParameter -ArgumentList "@MaxLSN", $binaryDataType

        $cdcChangesCommand.Parameters.Add($lastLSN) | Out-Null
        $cdcChangesCommand.Parameters.Add($maxLSN) | Out-Null

        $lastLSN.Value=$cdcTrackedTable.LastLSN
        $maxLSN.Value=$cdcMaxLSNCommand.ExecuteScalar()

        $cdcChangesReader=$cdcChangesCommand.ExecuteReader()

        try
        {
            while($cdcChangesReader.Read())
            {
                $changeRecord=New-Object -TypeName PSObject -Property @{ServerName=$serverName;
                                                              DatabaseName=$databaseName;
                                                              SchemaName=$cdcTrackedTable.SchemaName;
                                                              TableName=$cdcTrackedTable.TableName;}
                for($i=0;$i -lt $cdcChangesReader.FieldCount;$i++)
                {
                    $columnName=$cdcChangesReader.GetName($i)
                    $columnTrackedMatch=$tableTrackedColumns | Where-Object {$_.ColumnName -eq $columnName -or $columnName -eq ($_.ColumnName + "_" + $_.TargetReferenceColumnName) }
                    if($null -ne $columnTrackedMatch -or $columnName -like "_CDC_*")
                    {
                        $changeRecord | Add-Member $columnName $cdcChangesReader[$i]    
                    }
                    elseif($columnName -eq "__`$operation")
                    {
                        $operationDescription=""    
                        switch($cdcChangesReader[$i])
                        {
                            1 {$operationDescription="Delete"}
                            2 {$operationDescription="Insert"}
                            4 {$operationDescription="Update"}
                            default {$operationDescription="Unknown"}
                        }
                        $changeRecord | Add-Member "_CDC_OperationDescription" $operationDescription 
                    }
                }
                $changeRecord   
            }

            $cdcTrackedTable.LastLSN=$maxLSN.Value

            $cdcTrackedTable | Set-DataCollectorCDCTrackedTableLSN -environmentShortName $environmentShortName -dataCollectorConnection $dataCollectorConnection | Out-Null

            #Set-DataCollectorCDCTrackedTableLSN -sqlConnection $dataCollectorConnection -cdcTrackedTable $cdcTrackedTable -maxLSN ($maxLSN.Value)
        }
        catch
        {
            Write-Error -Message $_.Exception
        }
    }

    $dataCollectorConnection.Close()
    $cdcConnection.Close()

      
}

<#
.DESCRIPTION
    Gets the list of CDC enabled tables on a server in particular database.
.PARAMETER serverName
    The name of the server that contains the SQL Server instance that will be queried.
.PARAMETER databaseName
    The name of the database that will be queried.
.PARAMETER startWithMaxLSN
    Specifies whether the most current LSN should be returned as output instead of the first one.
.NOTES
    Author: Mathieu Isabel   
#>
Function Get-SQLCDCTable($serverName="",$databaseName="",$schemaName="",$tableName="",[switch]$startWithMAXLSN)
{
    Import-Module -DisableNameChecking -Force -Name ..\ConfigurationDatabase\libConfigurationDatabase.psm1
    Import-Module -DisableNameChecking -Force -Name ..\ServiceBus\libServiceBus.psm1

    $dataCollectorConnection = new-object -TypeName ("Data.SqlClient.SqlConnection") -ArgumentList "Data Source=$serverName;Initial Catalog=$databaseName;Integrated Security=True;MultipleActiveResultSets=True"

    if($dataCollectorConnection.State -ne "Open")
    {
        $dataCollectorConnection.Open()
    } 


    if($schemaName -eq "" -or $null -eq $schemaName -or 
        $tableName -eq "" -or $null -eq $tableName)
    {
	    $cdcTablesQuery = "EXEC sys.sp_cdc_help_change_data_capture"

        $cdcTablesCommand = New-Object -TypeName Data.SqlClient.SqlCommand -ArgumentList $cdcTablesQuery,$dataCollectorConnection
    }
    else
    {
        $cdcTablesQuery = "EXEC sys.sp_cdc_help_change_data_capture @source_schema=@SchemaName, @source_name=@TableName"

        $schemaNameParam = New-Object -TypeName System.Data.SqlClient.SqlParameter -ArgumentList "@SchemaName", $varcharDataType
        $tableNameParam = New-Object -TypeName System.Data.SqlClient.SqlParameter -ArgumentList "@TableName", $varcharDataType

        $cdcTablesCommand = New-Object -TypeName Data.SqlClient.SqlCommand -ArgumentList $cdcTablesQuery,$dataCollectorConnection

        $cdcTablesCommand.Parameters.Add($schemaNameParam) | Out-Null
        $cdcTablesCommand.Parameters.Add($tableNameParam) | Out-Null

        $schemaNameParam.Value=$schemaName
        $tableNameParam.Value=$tableName

    }
    

	

    $reader=$cdcTablesCommand.ExecuteReader()


    $cdcMaxLSNQuery="SELECT sys.fn_cdc_get_max_lsn()"
    $cdcMaxLSNCommand = New-Object -TypeName Data.SqlClient.SqlCommand -ArgumentList $cdcMaxLSNQuery,$dataCollectorConnection
    $maxLSN=$cdcMaxLSNCommand.ExecuteScalar()

    while($reader.Read())
    {
        if($startWithMAXLSN)
        {
            New-Object -TypeName PSObject -Property @{ServerName=$serverName;
                                            DatabaseName=$databaseName;
                                            SchemaName=$reader["source_schema"];
                                            TableName=$reader["source_table"];
                                            LastLSN=$maxLSN;
                                                                }    
        }
        else
        {
            New-Object -TypeName PSObject -Property @{ServerName=$serverName;
                                            DatabaseName=$databaseName;
                                            SchemaName=$reader["source_schema"];
                                            TableName=$reader["source_table"];
                                            LastLSN=$reader["start_lsn"];
                                                                }
        }
    }

    $dataCollectorConnection.Close()   
}

<#
.DESCRIPTION
    Adds a configuration record for a CDC tracked table in the DataCollector configuration database. 
    Expects an object from Get-SQLCDCTable on the pipeline.
.PARAMETER environmentShortName
    The short name of the environment that contains the configuration.
.NOTES
    Author: Mathieu Isabel   
#>
Function New-DataCollectorCDCTrackedTable($environmentShortName="",[ValidateSet('All','TrackedColumnsOnly')]$collectionType)
{
    Begin
    {
        Import-Module -DisableNameChecking -Force -Name ..\ConfigurationDatabase\libConfigurationDatabase.psm1
        Import-Module -DisableNameChecking -Force -Name ..\ServiceBus\libServiceBus.psm1

        $dataCollectorConnection = new-object -TypeName ("Data.SqlClient.SqlConnection") -ArgumentList (Get-ConfigurationSetting -environmentShortName $environmentShortName -scope "Global"  -name "DataCollectorDatabaseConnectionString").Value

        if($dataCollectorConnection.State -ne "Open")
        {
            $dataCollectorConnection.Open()
        }

        $varcharDataType=[System.Data.SqlDbType]::VarChar
        $binaryDataType=[System.Data.SqlDbType]::Binary
        

        $newDataCollectorCDCTrackedTableQuery="INSERT INTO [dbo].[CDCTrackedTables]
                                               ([ServerName]
                                               ,[DatabaseName]
                                               ,[SchemaName]
                                               ,[TableName]
                                               ,[LastLSN]
                                               ,[CollectionType])
                                         VALUES
                                               (@ServerName
                                               ,@DatabaseName
                                               ,@SchemaName
                                               ,@TableName
                                               ,@LastLSN
                                               ,@CollectionType)"


        $trackedDatabasesCommand = New-Object -TypeName Data.SqlClient.SqlCommand -ArgumentList $newDataCollectorCDCTrackedTableQuery,$dataCollectorConnection

        $serverName = New-Object -TypeName System.Data.SqlClient.SqlParameter -ArgumentList "@ServerName", $varcharDataType
        $databaseName = New-Object -TypeName System.Data.SqlClient.SqlParameter -ArgumentList "@DatabaseName", $varcharDataType
        $schemaName = New-Object -TypeName System.Data.SqlClient.SqlParameter -ArgumentList "@SchemaName", $varcharDataType
        $tableName = New-Object -TypeName System.Data.SqlClient.SqlParameter -ArgumentList "@TableName", $varcharDataType
        $lastLSN = New-Object -TypeName System.Data.SqlClient.SqlParameter -ArgumentList "@LastLSN", $binaryDataType
        $collectionTypeParam = New-Object -TypeName System.Data.SqlClient.SqlParameter -ArgumentList "@CollectionType", $varcharDataType

        $trackedDatabasesCommand.Parameters.Add($serverName) | Out-Null
        $trackedDatabasesCommand.Parameters.Add($databaseName) | Out-Null
        $trackedDatabasesCommand.Parameters.Add($schemaName) | Out-Null
        $trackedDatabasesCommand.Parameters.Add($tableName) | Out-Null
        $trackedDatabasesCommand.Parameters.Add($lastLSN) | Out-Null
        $trackedDatabasesCommand.Parameters.Add($collectionTypeParam) | Out-Null

    }

    Process
    {
        $serverName.Value=$_.ServerName
        $databaseName.Value=$_.DatabaseName
        $schemaName.Value=$_.SchemaName
        $tableName.Value=$_.TableName
        $lastLSN.Value=$_.StartLSN
        $collectionTypeParam.Value=$collectionType

        $trackedDatabasesCommand.ExecuteNonQuery()
    }

    End
    {
        $dataCollectorConnection.Close()
    }
}

<#
.DESCRIPTION
    Resets a configuration record for a CDC tracked table in the DataCollector configuration database. 
    Expects an object from Get-SQLCDCTable on the pipeline.
.PARAMETER environmentShortName
    The short name of the environment that contains the configuration.
.NOTES
    Author: Mathieu Isabel   
#>
Function Set-DataCollectorCDCTrackedTableLSN($environmentShortName="",$dataCollectorConnection=$null)
{
    Begin
    {
        Import-Module -DisableNameChecking -Force -Name ..\ConfigurationDatabase\libConfigurationDatabase.psm1
        Import-Module -DisableNameChecking -Force -Name ..\ServiceBus\libServiceBus.psm1

        if($null -eq $dataCollectorConnection)
        {
            $dataCollectorConnection = new-object -TypeName ("Data.SqlClient.SqlConnection") -ArgumentList (Get-ConfigurationSetting -environmentShortName $environmentShortName -scope "Global"  -name "DataCollectorDatabaseConnectionString").Value
        }

        if($dataCollectorConnection.State -ne "Open")
        {
            $dataCollectorConnection.Open()
        }

        $varcharDataType=[System.Data.SqlDbType]::VarChar
        $binaryDataType=[System.Data.SqlDbType]::Binary
        

        $newDataCollectorCDCTrackedTableQuery="UPDATE [dbo].[CDCTrackedTables]
                                                   SET [LastLSN]=@LastLSN
                                               WHERE [ServerName]=@ServerName AND 
                                                   [DatabaseName]=@DatabaseName AND 
                                                   [SchemaName]=@SchemaName AND 
                                                   [TableName]=@TableName"


        $trackedDatabasesCommand = New-Object -TypeName Data.SqlClient.SqlCommand -ArgumentList $newDataCollectorCDCTrackedTableQuery,$dataCollectorConnection

        $serverName = New-Object -TypeName System.Data.SqlClient.SqlParameter -ArgumentList "@ServerName", $varcharDataType
        $databaseName = New-Object -TypeName System.Data.SqlClient.SqlParameter -ArgumentList "@DatabaseName", $varcharDataType
        $schemaName = New-Object -TypeName System.Data.SqlClient.SqlParameter -ArgumentList "@SchemaName", $varcharDataType
        $tableName = New-Object -TypeName System.Data.SqlClient.SqlParameter -ArgumentList "@TableName", $varcharDataType
        $lastLSN = New-Object -TypeName System.Data.SqlClient.SqlParameter -ArgumentList "@LastLSN", $binaryDataType

        $trackedDatabasesCommand.Parameters.Add($serverName) | Out-Null
        $trackedDatabasesCommand.Parameters.Add($databaseName) | Out-Null
        $trackedDatabasesCommand.Parameters.Add($schemaName) | Out-Null
        $trackedDatabasesCommand.Parameters.Add($tableName) | Out-Null
        $trackedDatabasesCommand.Parameters.Add($lastLSN) | Out-Null

    }

    Process
    {
        $serverName.Value=$_.ServerName
        $databaseName.Value=$_.DatabaseName
        $schemaName.Value=$_.SchemaName
        $tableName.Value=$_.TableName
        $lastLSN.Value=$_.LastLSN

        $trackedDatabasesCommand.ExecuteNonQuery() | Out-Null
    }

    End
    {
        $dataCollectorConnection.Close()
    }    
}

<#
.DESCRIPTION
    Gets the list of servers and databases that are currently tracked by the DataCollector.
.PARAMETER environmentShortName
    The short name of the environment that contains the configuration.
.NOTES
    Author: Mathieu Isabel   
#>
Function Get-DataCollectorCDCTrackedDatabase($environmentShortName="")
{
    Import-Module -DisableNameChecking -Force -Name ..\ConfigurationDatabase\libConfigurationDatabase.psm1
    Import-Module -DisableNameChecking -Force -Name ..\ServiceBus\libServiceBus.psm1

    $dataCollectorConnection = new-object -TypeName ("Data.SqlClient.SqlConnection") -ArgumentList (Get-ConfigurationSetting -environmentShortName $environmentShortName -scope "Global"  -name "DataCollectorDatabaseConnectionString").Value

    if($dataCollectorConnection.State -ne "Open")
    {
        $dataCollectorConnection.Open()
    } 


	$trackedDatabasesQuery = "SELECT DISTINCT ServerName,DatabaseName 
                                FROM CDCTrackedTables 
                                ORDER BY ServerName,DatabaseName";
	$trackedDatabasesCommand = New-Object -TypeName Data.SqlClient.SqlCommand -ArgumentList $trackedDatabasesQuery,$dataCollectorConnection

    $reader=$trackedDatabasesCommand.ExecuteReader()

    $cdcTrackedDatabases=@()

    while($reader.Read())
    {
        $cdcTrackedDatabases+=New-Object -TypeName PSObject -Property @{ServerName=$reader["ServerName"];
                                                            DatabaseName=$reader["DatabaseName"];
                                                            }
    }

    $dataCollectorConnection.Close()

    $cdcTrackedDatabases
}

<#
.DESCRIPTION
    Collects the CDC change records on the configured servers, databases, tables and columns.
.PARAMETER environmentShortName
    The short name of the environment that contains the configuration.
.NOTES
    Author: Mathieu Isabel   
#>
Function Get-DataCollectorCDCChange($environmentShortName="")
{
    Import-Module -DisableNameChecking -Force -Name ..\ConfigurationDatabase\libConfigurationDatabase.psm1

    $collectionIntervalInSeconds=(Get-ConfigurationSetting -environmentShortName $environmentShortName -scope DataCollector  -name "DataCollectorCollectionIntervalInSeconds").Value

    $trackedDatabases=Get-DataCollectorCDCTrackedDatabase

    while($true)
    {
        foreach($trackedDatabase in $trackedDatabases)
        {
            Write-Host "Collecting changes for"$trackedDatabase.ServerName $trackedDatabase.DatabaseName 
            $cdcTrackedTables=Get-DataCollectorCDCTrackedTable -serverName $trackedDatabase.ServerName -databaseName $trackedDatabase.DatabaseName
            Get-DataCollectorCDCDatabaseChange -environmentShortName $environmentShortName -serverName $trackedDatabase.ServerName -databaseName $trackedDatabase.DatabaseName -cdcTrackedTables $cdcTrackedTables  
        }
        Write-Host "Pausing $collectionIntervalInSeconds seconds before the next collection"
        Start-Sleep -Seconds $collectionIntervalInSeconds
    }
}

<#
.DESCRIPTION
    Gets the list of CDC tables tracked by the DataCollector on a specific server and database.
.PARAMETER environmentShortName
    The short name of the environment that contains the configuration.
.PARAMETER serverName
    The name of the server that contains the SQL Server instance for which the changes will be collected.
.PARAMETER databaseName
    The name of the database for which the changes will be collected.
.NOTES
    Author: Mathieu Isabel   
#>
Function Get-DataCollectorCDCTrackedTable($environmentShortName="",$serverName="",$databaseName="")
{
    Import-Module -DisableNameChecking -Force -Name ..\ConfigurationDatabase\libConfigurationDatabase.psm1
    Import-Module -DisableNameChecking -Force -Name ..\ServiceBus\libServiceBus.psm1

    $dataCollectorConnection = new-object -TypeName ("Data.SqlClient.SqlConnection") -ArgumentList (Get-ConfigurationSetting -environmentShortName $environmentShortName -scope "Global"  -name "DataCollectorDatabaseConnectionString").Value

    if($dataCollectorConnection.State -ne "Open")
    {
        $dataCollectorConnection.Open()
    } 

    $varcharDataType=[System.Data.SqlDbType]::VarChar
    $dateTimeDataType=[System.Data.SqlDbType]::DateTime
    $smallIntDataType=[System.Data.SqlDbType]::SmallInt
    $intDataType=[System.Data.SqlDbType]::Int
    $bigIntDataType=[System.Data.SqlDbType]::BigInt

    $trackedTablesQuery="SELECT SchemaName,
                                TableName,
                                LastLSN,
                                CollectionType
                        FROM CDCTrackedTables
                        WHERE ServerName = @ServerName AND
                              DatabaseName = @DatabaseName AND
                              CollectionEnabled = 1"


    $trackedTablesCommand = New-Object -TypeName Data.SqlClient.SqlCommand -ArgumentList $trackedTablesQuery,$dataCollectorConnection

    $serverNameParam = New-Object -TypeName System.Data.SqlClient.SqlParameter -ArgumentList "@ServerName", $varcharDataType
    $databaseNameParam = New-Object -TypeName System.Data.SqlClient.SqlParameter -ArgumentList "@DatabaseName", $varcharDataType

    $trackedTablesCommand.Parameters.Add($serverNameParam) | Out-Null
    $trackedTablesCommand.Parameters.Add($databaseNameParam) | Out-Null

    $serverNameParam.Value=$serverName
    $databaseNameParam.Value=$databaseName

    $reader=$trackedTablesCommand.ExecuteReader()


    while($reader.Read())
    {
        New-Object -TypeName PSObject -Property @{ServerName=$serverName;
                                                DatabaseName=$databaseName;
                                                SchemaName=$reader["SchemaName"];
                                                TableName=$reader["TableName"];
                                                LastLSN=$reader["LastLSN"];
                                                CollectionType=$reader["CollectionType"];}
    }

    $dataCollectorConnection.Close() 
}

<#
.DESCRIPTION
    Gets the list of columns that are tracked by the DataCollector on a specific table.
.PARAMETER environmentShortName
    The short name of the environment that contains the configuration.
.PARAMETER serverName
    The name of the server that contains the SQL Server instance for which the changes will be collected.
.PARAMETER databaseName
    The name of the database for which the changes will be collected.
.PARAMETER schemaName
    The name of the schema that contains the tables for which the changes will be collected.
.PARAMETER tableName
    The name of the column that is targeted by the DataCollector collection.
.NOTES
    Author: Mathieu Isabel   
#>
Function Get-DataCollectorCDCTrackedTableColumn($environmentShortName="",$serverName="",$databaseName="",$schemaName="",$tableName="")
{
    Import-Module -DisableNameChecking -Force -Name ..\ConfigurationDatabase\libConfigurationDatabase.psm1
    Import-Module -DisableNameChecking -Force -Name ..\ServiceBus\libServiceBus.psm1

    $dataCollectorConnection = new-object -TypeName ("Data.SqlClient.SqlConnection") -ArgumentList (Get-ConfigurationSetting -environmentShortName $environmentShortName -scope "Global"  -name "DataCollectorDatabaseConnectionString").Value

    if($dataCollectorConnection.State -ne "Open")
    {
        $dataCollectorConnection.Open()
    } 

    $varcharDataType=[System.Data.SqlDbType]::VarChar


    $trackedTableColumnsQuery="SELECT ColumnName,
                                      TargetTableName,
                                      TargetKeyColumnName,
                                      TargetReferenceColumnName
                                FROM CDCTrackedTableColumns
                                WHERE ServerName = @ServerName AND
                                      DatabaseName = @DatabaseName AND
                                      SchemaName= @SchemaName AND
                                      TableName= @TableName"


    $trackedTableColumnsCommand = New-Object -TypeName Data.SqlClient.SqlCommand -ArgumentList $trackedTableColumnsQuery,$dataCollectorConnection

    $serverNameParam = New-Object -TypeName System.Data.SqlClient.SqlParameter -ArgumentList "@ServerName", $varcharDataType
    $databaseNameParam = New-Object -TypeName System.Data.SqlClient.SqlParameter -ArgumentList "@DatabaseName", $varcharDataType
    $schemaNameParam = New-Object -TypeName System.Data.SqlClient.SqlParameter -ArgumentList "@SchemaName", $varcharDataType
    $tableNameParam = New-Object -TypeName System.Data.SqlClient.SqlParameter -ArgumentList "@TableName", $varcharDataType

    $trackedTableColumnsCommand.Parameters.Add($serverNameParam) | Out-Null
    $trackedTableColumnsCommand.Parameters.Add($databaseNameParam) | Out-Null
    $trackedTableColumnsCommand.Parameters.Add($schemaNameParam) | Out-Null
    $trackedTableColumnsCommand.Parameters.Add($tableNameParam) | Out-Null

    $serverNameParam.Value=$serverName
    $databaseNameParam.Value=$databaseName
    $schemaNameParam.Value=$schemaName
    $tableNameParam.Value=$tableName


    $reader=$trackedTableColumnsCommand.ExecuteReader()


    while($reader.Read())
    {
        New-Object -TypeName PSObject -Property @{ServerName=$serverName;
                                        DatabaseName=$databaseName;
                                        SchemaName=$schemaName;
                                        TableName=$tableName;
                                        ColumnName=$reader["ColumnName"];
                                        TargetTableName=$reader["TargetTableName"];
                                        TargetKeyColumnName=$reader["TargetKeyColumnName"];
                                        TargetReferenceColumnName=$reader["TargetReferenceColumnName"];}
    }

    $dataCollectorConnection.Close() 
}

<#
.DESCRIPTION
    Creates a new column that will for which CDC changes will be collected by the DataCollector.
.PARAMETER environmentShortName
    The short name of the environment that contains the configuration.
.PARAMETER serverName
    The name of the server that contains the SQL Server instance for which the changes will be collected.
.PARAMETER databaseName
    The name of the database for which the changes will be collected.
.PARAMETER schemaName
    The name of the schema that contains the tables for which the changes will be collected.
.PARAMETER tableName
    The name of the table for which the changes will be collected.
.PARAMETER columnName
    The name of the column that will be collected.
.NOTES
    Author: Mathieu Isabel   
#>
Function New-DataCollectorCDCTrackedTableColumn($environmentShortName="",$serverName="",$databaseName="",$schemaName="",$tableName="",$columnName="")
{
    Import-Module -DisableNameChecking -Force -Name ..\ConfigurationDatabase\libConfigurationDatabase.psm1
    Import-Module -DisableNameChecking -Force -Name ..\ServiceBus\libServiceBus.psm1

    $dataCollectorConnection = new-object -TypeName ("Data.SqlClient.SqlConnection") -ArgumentList (Get-ConfigurationSetting -environmentShortName $environmentShortName -scope "Global"  -name "DataCollectorDatabaseConnectionString").Value

    if($dataCollectorConnection.State -ne "Open")
    {
        $dataCollectorConnection.Open()
    } 

    $varcharDataType=[System.Data.SqlDbType]::VarChar
    $dateTimeDataType=[System.Data.SqlDbType]::DateTime
    $smallIntDataType=[System.Data.SqlDbType]::SmallInt
    $intDataType=[System.Data.SqlDbType]::Int
    $bigIntDataType=[System.Data.SqlDbType]::BigInt

    $trackedTableColumnsQuery="INSERT INTO [dbo].[CDCTrackedTableColumns]
                               ([ServerName]
                               ,[DatabaseName]
                               ,[SchemaName]
                               ,[TableName]
                               ,[ColumnName])
                         VALUES
                               (@ServerName
                               ,@DatabaseName
                               ,@SchemaName
                               ,@TableName
                               ,@ColumnName)"


    $trackedTableColumnsCommand = New-Object -TypeName Data.SqlClient.SqlCommand -ArgumentList $trackedTableColumnsQuery,$dataCollectorConnection

    $serverNameParam = New-Object -TypeName System.Data.SqlClient.SqlParameter -ArgumentList "@ServerName", $varcharDataType
    $databaseNameParam = New-Object -TypeName System.Data.SqlClient.SqlParameter -ArgumentList "@DatabaseName", $varcharDataType
    $schemaNameParam = New-Object -TypeName System.Data.SqlClient.SqlParameter -ArgumentList "@SchemaName", $varcharDataType
    $tableNameParam = New-Object -TypeName System.Data.SqlClient.SqlParameter -ArgumentList "@TableName", $varcharDataType
    $columnNameParam = New-Object -TypeName System.Data.SqlClient.SqlParameter -ArgumentList "@ColumnName", $varcharDataType

    $trackedTableColumnsCommand.Parameters.Add($serverNameParam) | Out-Null
    $trackedTableColumnsCommand.Parameters.Add($databaseNameParam) | Out-Null
    $trackedTableColumnsCommand.Parameters.Add($schemaNameParam) | Out-Null
    $trackedTableColumnsCommand.Parameters.Add($tableNameParam) | Out-Null
    $trackedTableColumnsCommand.Parameters.Add($columnNameParam) | Out-Null

    $serverNameParam.Value=$serverName
    $databaseNameParam.Value=$databaseName
    $schemaNameParam.Value=$schemaName
    $tableNameParam.Value=$tableName
    $columnNameParam.Value=$columnName


    $trackedTableColumnsCommand.ExecuteNonQuery()

    $dataCollectorConnection.Close() 
}

<#
.DESCRIPTION
    Converts CDC event to a hash table which can then be piped in something like Send-SBEventHubMessage
.NOTES
    Author: Mathieu Isabel   
#>
Function Convert-DataCollectorCDCChangeToHashTable()
{
    Process
    {
        $output=@{}
        $properties=$_ | Get-Member | Where-Object {$_.MemberType -eq "NoteProperty"} | Sort-Object -Property Name
        foreach($property in $properties)
        {
            $output.Add($property.Name,$_.($property.Name))   
        }
        $output  
    }
}

<#
.DESCRIPTION
    Generates the query that fetches foreign key descriptive values.
.PARAMETER dataCollectorTrackedColumns
    The columns that are captured by the DataCollector
.NOTES
    Author: Mathieu Isabel   
#>
Function Get-DataCollectorCDCQuery($dataCollectorTrackedTable=$null,$dataCollectorTrackedColumns=$null)
{
    $cdcChangesQuery="SELECT "
                            
    $baseColumns=$dataCollectorTrackedColumns  | Select-Object ColumnName -Unique
    $referenceColumns=$dataCollectorTrackedColumns | Where-Object {$_.TargetTableName -ne [System.DBNull]::Value}| Select-Object ColumnName,TargetTableName,TargetReferenceColumnName -Unique
    
    foreach($column in $baseColumns)
    {
        $cdcChangesQuery+="cdc." + $column.ColumnName + ", `r`n"     
    }

    foreach($column in $referenceColumns)
    {
        $cdcChangesQuery+=$column.TargetTableName + "_" + $column.ColumnName + "." + $column.TargetReferenceColumnName + " AS " + $column.ColumnName + "_" + $column.TargetReferenceColumnName + ", `r`n"     
    }

    $cdcChangesQuery+="sys.fn_cdc_map_lsn_to_time(__`$start_lsn) AS _CDC_ChangeTime, `r`n"
    $cdcChangesQuery+="cdc.__`$operation `r`n"
    $cdcChangesQuery+="FROM  [cdc].[fn_cdc_get_all_changes_" + $dataCollectorTrackedTable.SchemaName + "_" + $dataCollectorTrackedTable.TableName + "](@LastLSN, @MaxLSN,N'all') AS cdc `r`n"

    $joins=$dataCollectorTrackedColumns | Where-Object {$_.TargetTableName -ne [System.DBNull]::Value} | Select-Object -Property ServerName,DatabaseName,ColumnName,TargetTableName,TargetKeyColumnName -Unique 

    foreach($join in $joins)
    {
        if($join.TargetTableName -ne [System.DBNull]::Value)
        {
            $cdcChangesQuery+="INNER JOIN " + $join.TargetTableName + " AS "+ $join.TargetTableName + "_" + $join.ColumnName +" ON "  + "cdc." + $join.ColumnName + "=" + $join.TargetTableName + "_" + $join.ColumnName +"." + $join.TargetKeyColumnName + "`r`n"
        }
    }


    
    $cdcChangesQuery+="WHERE [__`$start_lsn] !=@LastLSN"

    if($dataCollectorTrackedTable.CollectionType -eq "TrackedColumnsOnly")
    {
        $cdcChangesQuery+=" AND (`r`n"
        for($i=0;$i -lt $baseColumns.Count;$i++)
        {
            $cdcChangesQuery+=" sys.fn_cdc_is_bit_set (sys.fn_cdc_get_column_ordinal ('"+ $dataCollectorTrackedTable.SchemaName +"_" + $dataCollectorTrackedTable.TableName + "','"+ $baseColumns[$i].ColumnName +"'),__`$update_mask) = 1 "
            if($i+1 -lt $baseColumns.Count)
            {
                $cdcChangesQuery+=" OR `r`n"
            }

        }
        $cdcChangesQuery+=")"        
    }

    $cdcChangesQuery 
}

<#
.DESCRIPTION
    Resets the LSN for the tracked tables to the first or last LSN possible.
.PARAMETER environmentShortName
    The environment short name that contains the configuration for the data collector
.PARAMETER startWithMaxLSN
    If specified, the last LSN reported by CDC will be used as the last point captured.If omitted, the first will be used.
.NOTES
    Author: Mathieu Isabel   
#>
Function Reset-DataCollectorCDCTrackedTableLSN($environmentShortName=$null,$serverName="",$databaseName="",$schemaName="",$tableName="",[switch]$startWithMAXLSN=$false)
{
    if($serverName -eq "" -and $databaseName -eq "")
    {
        $cdcTrackedDatabases=Get-DataCollectorCDCTrackedDatabase -environmentShortName $environmentShortName
    }
    else
    {
        $cdcTrackedDatabases=Get-DataCollectorCDCTrackedDatabase -environmentShortName $environmentShortName | Where-Object {$_.ServerName -eq $serverName -and $_.DatabaseName -eq $databaseName}
    }

    foreach($cdcTrackedDatabase in $cdcTrackedDatabases)
    {
        
        if($tableName -eq "" -and $schemaName -eq "")
        {
            $cdcTrackedTables=Get-DataCollectorCDCTrackedTable -environmentShortName $environmentShortName -serverName $cdcTrackedDatabase.ServerName -databaseName $cdcTrackedDatabase.DatabaseName
        }
        else
        {
            $cdcTrackedTables=Get-DataCollectorCDCTrackedTable -environmentShortName $environmentShortName -serverName $cdcTrackedDatabase.ServerName -databaseName $cdcTrackedDatabase.DatabaseName | Where-Object {$_.SchemaName -eq $schemaName -and $_.TableName -eq $tableName}
        }
        
        foreach($cdcTrackedTable in $cdcTrackedTables)
        {
            $message="Resetting " + $cdcTrackedTable.ServerName + "." + $cdcTrackedTable.DatabaseName + "." + $cdcTrackedTable.SchemaName + "." + $cdcTrackedTable.TableName
            Write-Output $message
            Get-SQLCDCTable -serverName $cdcTrackedTable.ServerName -databaseName $cdcTrackedTable.DatabaseName -startWithMAXLSN:$startWithMAXLSN.IsPresent -schemaName $cdcTrackedTable.SchemaName -tableName $cdcTrackedTable.TableName | Set-DataCollectorCDCTrackedTableLSN -environmentShortName $environmentShortName
        } 
    }
}